package com.study.mq.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author ysy
 * @version 1.0
 * @Title: KafkaProducer
 * @Note <b>ProjectName:</b> Pre-research
 * <br><b>PackageName:</b> com.study.mq.kafka
 * <br><b>ClassName:</b>
 * <br><b>Date:</b>
 */
@RestController
@RequestMapping("/kafka")
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> template;

    @GetMapping("/send")
    String send(String topic, String data) {
        template.send(topic, data);
        return "success";
    }

    /**
     * @Author ysy
     * @Description 发送消息，且等待broke响应，直至发送成功，超时则异常处理
     * @Date 17:30 [topic, data]
     * @Param 
     * @return java.lang.String
     **/
    @GetMapping("/sendWaitResult")
    String sendWaitResult(String topic, String data) {
        String result = "发送成功";
        ListenableFuture<SendResult<String, String>> future = template.send(topic, data);
        try {
            future.get(5, TimeUnit.SECONDS);
        }catch (InterruptedException e){
            e.printStackTrace();
            result = "发送失败";
        }catch (ExecutionException e){
            e.printStackTrace();
            result = "执行失败";
        }catch (TimeoutException e){
            e.printStackTrace();
            result = "发送超时";
        }
        return result;
    }

}
